package org.jeecg.modules.iot.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jeecg.common.api.vo.Result;
import org.jeecg.jt808.vo.ClientConnectEvent;
import org.jeecg.modules.iot.mqtt.server.MqttSendClient;
import org.jeecg.modules.iot.mqtt.server.service.processor.carOperate.CarService;
import org.jeecg.modules.rental.api.IRentalEBikeAPI;
import org.jeecg.modules.rental.entity.IotTerminalTemp;
import org.jeecg.modules.rental.entity.RentalEBike;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
public class KafkaToMqttForwarder {

    private static final Logger logger = LoggerFactory.getLogger(KafkaToMqttForwarder.class);

    @Autowired
    private MqttSendClient mqttSendClient; //  注入 MqttSendClient

    @Autowired
    private IRentalEBikeAPI iRentalEBikeAPI;

    private final ObjectMapper objectMapper = new ObjectMapper();


    @KafkaListener(topics = "device_send", groupId = "mqtt-forwarder", concurrency = "3") //  根据实际情况修改 concurrency
    public void listen(ConsumerRecord<String, String> record) {
        try {
            String key = record.key(); // 获取消息的 key，即 carCode
            String message = record.value(); // 获取消息的值，即 JT/T 808 指令的 JSON 字符串
            System.out.println("接收到"+key);
            // 发送消息到 MQTT
            mqttSendClient.publish(true, key, message); // 使用 carCode 作为 topic

        } catch (Exception e) {
            log.error("转发 Kafka 消息到 MQTT 失败", e);
            // ... 错误处理 ...
        }
    }

    @KafkaListener(topics = "device-status-topic", groupId = "mqtt-forwarder", concurrency = "3") //  根据实际情况修改 concurrency
    public void listenStatus(ConsumerRecord<String, String> record) {
        try {
            String key = record.key(); // 获取消息的 key，即 carCode
            String message = record.value(); // 获取消息的值，即 JT/T 808 指令的 JSON 字符串
            System.out.println("接收到"+key);
            ClientConnectEvent clientConnectEvent = objectMapper.readValue(message, ClientConnectEvent.class);
            IotTerminalTemp terminalTemp=new IotTerminalTemp();
            terminalTemp.setSn(clientConnectEvent.getClientid());
            terminalTemp.setImei(clientConnectEvent.getClientid());
            RentalEBike rentalEBike=new RentalEBike();
            rentalEBike.setCarCode(clientConnectEvent.getClientid());
            Result<List<RentalEBike>> result = iRentalEBikeAPI.listIotcar(rentalEBike);
            if (result.isSuccess() && !result.getResult().isEmpty()) {
                return;
            }
            Result<List<IotTerminalTemp>> equTerminalsResult = iRentalEBikeAPI.listIot(terminalTemp);
            if (equTerminalsResult.isSuccess() && equTerminalsResult.getResult().isEmpty()) { // Check for success and empty list
                Result<String> addResult = iRentalEBikeAPI.addTerminal(terminalTemp);
                if (!addResult.isSuccess()) {
                    log.error("Failed to add EquTerminal: {}", addResult.getMessage());
                }

                 }

        } catch (Exception e) {
            log.error("添加设备失败", e);
            // ... 错误处理 ...
        }
    }



}